package com.zmp.kotlinnote.coroutines

import com.zmp.kotlinnote.exchange.log
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import kotlinx.coroutines.selects.select
import org.junit.Test
import kotlin.system.measureTimeMillis

/**
 *@author zmp
 *Created by zmp on 2018/12/25
 *TODO
 */
class _XC_3 {
    private val channel = Channel<Int>()
    @Test
    fun launchVoid() = runBlocking<Unit> {
        launch {
            repeat(5) {
                delay(300)
                "channel.send $it".log()
                channel.send(it)
            }
            channel.close()//关闭通道
            "channel -- close".log()
        }
        "launch  -- send".log()
        launch {
            for (i in channel) { //通道未关闭 channel会阻塞
                delay(1000)
                i.log()
            }
        }
        "launch  -- receive".log()
    }


    @ExperimentalCoroutinesApi
    val produce = GlobalScope.produce {
        repeat(5) {
            send("produce send $it")
            delay(1000)
        }
        close()
    }.map {
        "filter $it"
    }

    @ObsoleteCoroutinesApi
    @ExperimentalCoroutinesApi
    @Test
    fun launchVoid2() = runBlocking<Unit> {
        produce.consumeEach {
            it.log()
        }
//        delay(3000)
//        coroutineContext.cancelChildren()
    }

    @Test
    fun measureTime() {
        try {
            measureTimeMillis {
                runBlocking {
                    val async = async(start = CoroutineStart.LAZY) {
                        try {
                            delay(2000)
                            "async one"
                        } catch (run: CancellationException) {
                            "catch RuntimeException".log()
                            run.printStackTrace()
                        } finally {
                            "async one RuntimeException".log()  // 函数内部发生了错误，并且它抛出了一个JobCancellationException异常， 所有在作用域中启动的协程都将会被取消
                        }
                    }

                    val async2 = async(start = CoroutineStart.LAZY) {
                        delay(1000)
                        "async two".log()
                        throw ArithmeticException("async two error")
                    }
                    async.start()
                    async2.start()
                    async.await()
                    async2.await()
                }
            }.log()
        } catch (run: RuntimeException) {
            run.printStackTrace()
        } finally {
            "measureTimeMillis finally".log()
        }
    }

    @Test
    fun main() = runBlocking<Unit> {
        //捕获异常
        val handler = CoroutineExceptionHandler { _, exception ->
            println("Caught $exception --${exception.suppressed?.contentToString()}")
        }
        val job = GlobalScope.launch(handler) {
            println("Throwing exception from launch")
            throw ArithmeticException() // Will be printed to the console by Thread.defaultUncaughtExceptionHandler
        }
        job.join()
        val deferred = GlobalScope.async(handler) {
            println("Throwing exception from async")
            throw ArithmeticException() //没有打印任何东西，async 依赖用户去调用 deferred.await()
        }

        try {
            deferred.await() //会抛出异常
        } catch (e: Exception) {
            e.printStackTrace()
        }

        GlobalScope.launch(context = handler) {

            launch {
                delay(500)
                throw RuntimeException("RuntimeException  1111")
            }

            launch {
                delay(300)
                throw RuntimeException("RuntimeException  2222")
            }

            launch {
                delay(100)
                throw RuntimeException("RuntimeException  333")
            }
        }.join()

    }


    @ExperimentalCoroutinesApi
    @Test
    fun main2() = runBlocking<Unit> {
        val channel1 = Channel<Deferred<String>>()
        val channel2 = Channel<Deferred<String>>()
        launch {
            while (isActive) {
                "isActive  -----".log()
                val a = select<String> {
                    channel1.onReceive {
                        "channel1  --".log()
                        delay(1000)
                        it.await()
                    }
                    channel2.onReceive {
                        "channel2  --".log()
                        it.await()
                    }
                }
                a.log()
            }
        }

        channel2.send(asyncStringAsync("channel  2", 1000))
        channel2.send(asyncStringAsync("channel  2", 1000))
        channel2.send(asyncStringAsync("channel  2", 1000))
        channel2.send(asyncStringAsync("channel  2", 1000))
        channel1.send(asyncStringAsync("channel  1", 2000))
        channel1.close()
        channel2.close()
        coroutineContext.cancelChildren()

    }

    @ObsoleteCoroutinesApi
    @ExperimentalCoroutinesApi
    fun CoroutineScope.switchMapDeferreds(input: ReceiveChannel<Deferred<String>>) = produce<String> {
        var current = input.receive() // start with first received deferred value
        while (isActive) { // loop while not cancelled/closed
            val next = select<Deferred<String>?> {
                // return next deferred value from this select or null
                input.onReceiveOrNull { update ->
                    "update".log()
                    update // replaces next value to wait
                }
                current.onAwait { value ->
                    value.log()
                    send(value) // send value that current deferred has produced
                    input.receiveOrNull() // and use the next deferred from the input channel
                }
            }
            if (next == null) {
                println("Channel was closed")
                break // out of loop
            } else {
                current = next
            }
        }
    }

    private fun CoroutineScope.asyncStringAsync(str: String, time: Long) = async {
        delay(time)
        str
    }

}